package com.example.service;


import com.example.utils.DataChange;
import org.jgroups.*;
import org.jgroups.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import java.io.InputStream;
import java.io.OutputStream;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/****
 *
 * 基于jgroups的集群消息推送
 */
public class JgroupsClusterServiceImpl implements ClusterService, Receiver {

    public static final String DEFAULT = "default";
    private final Map<String, String> pushMap = new ConcurrentHashMap<>();
    Logger logger = LoggerFactory.getLogger(getClass());
    private ApplicationContext applicationContext;
    private IndexService indexService;
    private UUID uuid;
    private JChannel channel;

    @Override
    public void start(ApplicationContext applicationContext, IndexService indexService) {
        try {
            this.applicationContext = applicationContext;
            this.indexService = indexService;
            this.channel = new JChannel("jgroups/jgroups-udp.xml");
            this.channel.setReceiver(this);
            this.channel.connect(DEFAULT);
            this.uuid = (UUID) this.channel.getAddress();
        } catch (Exception e) {
            logger.warn("jgroups server 启动失败 ...");
            if (this.channel != null) {
                this.channel.disconnect();
                this.channel.close();
            }
        }
    }

    @Override
    public void stop() {
        if (this.channel != null) {
            this.channel.disconnect();
            this.channel.close();
        }
    }

    @Override
    public void notifyToMessage(DataChange dataChange) {
        try {
            this.pushMap.put(dataChange.getId(), dataChange.getId());
            this.channel.send(null, dataChange);
        } catch (Exception e) {
            logger.warn("推送集群消息失败: " + e.getMessage());
        }
    }

    @Override
    public void viewAccepted(View new_view) {

    }

    @Override
    public void suspect(Address suspected_mbr) {

    }

    @Override
    public void block() {

    }

    @Override
    public void unblock() {

    }

    @Override
    public void receive(Message msg) {
        DataChange dataChange = (DataChange) msg.getObject();
        if (!this.pushMap.containsKey(dataChange.getId())) {
            indexService.onMessage(dataChange);
        }
    }

    @Override
    public void getState(OutputStream output) throws Exception {

    }

    @Override
    public void setState(InputStream input) throws Exception {

    }

    public Map<String, String> getPushMap() {
        return pushMap;
    }

    public ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    public IndexService getIndexService() {
        return indexService;
    }

    public void setIndexService(IndexService indexService) {
        this.indexService = indexService;
    }

    public UUID getUuid() {
        return uuid;
    }

    public void setUuid(UUID uuid) {
        this.uuid = uuid;
    }

    public JChannel getChannel() {
        return channel;
    }

    public void setChannel(JChannel channel) {
        this.channel = channel;
    }
}
